package com.infoauto.opc.gateway.messages.handler;


import com.infoauto.opc.gateway.messages.model.MqttAccess;
import com.infoauto.opc.gateway.messages.model.MqttConsumer;
import com.infoauto.opc.gateway.messages.util.SerializeUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//MQTT回调
public class SimpleMqttCallback implements MqttCallback {

    private static Logger logger = LoggerFactory.getLogger(SimpleMqttCallback.class);

    private MqttListener mqttListener;

    private MqttClient mqttClient;

    private MqttAccess mqttAccess;

    private Class genericType;

    public SimpleMqttCallback(MqttListener mqttListener, MqttClient mqttClient, MqttAccess mqttAccess) {
        this.mqttListener = mqttListener;
        this.mqttClient = mqttClient;
        this.mqttAccess = mqttAccess;
        this.genericType = mqttListener.getGenericType();
    }

    @Override
    public void connectionLost(Throwable cause) {
        logger.error(cause.getMessage(), cause);
        //丢失连接
        try {
            //重新连接
            mqttClient.connect();
            //重新订阅
            if (mqttAccess instanceof MqttConsumer) {
                MqttConsumer mqttConsumer = (MqttConsumer) mqttAccess;
                mqttConsumer.reset();
            }
        } catch (MqttException e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        mqttListener.messageArrived(topic, SerializeUtils.deserialize(message.getPayload(), genericType));
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        String[] topic = token.getTopics();
        if (topic != null && topic.length > 0) {
            mqttListener.deliveryComplete(topic[0]);
        }
    }
}
